package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Code32BroadCast {
  def main(args: Array[String]): Unit = {

    /**
     *
     */

    val num = 0;
    val conf: SparkConf = new SparkConf().setMaster("local").setAppName("Cache")
    val sc = new SparkContext(conf)
    sc.setCheckpointDir("spark_code/output/checkpoint")


    val wordsRDD: RDD[String] = sc.parallelize(
      List("hello spark", "hello scala", "hello world")
    )

    val map: Map[String, Int] = Map(("hello", 1), ("spark", 2), ("scala", 3), ("world", 4))
    val mapBroadcast: Broadcast[Map[String, Int]] = sc.broadcast(map)


    wordsRDD.flatMap(
      x => {
        println("flatMap")
        x.split(" ")
      }
    )
      .map(
        word => {
          // 广播变量的值进行获取
          val getWord: Int = mapBroadcast.value.getOrElse(word, 0)

//          val getWord: Int = map.getOrElse(word, 0)

          (word, getWord + 1)
        }
      ).foreach(println)

//    while (true) {}
  }
}
